#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <string>
#include <vector>
#include <iostream>
#include <stack>
#include <algorithm>
#include <map>
#include <cassert>

#include "../include/client.h"
#include "../lib/tokenize.h"
#include "../lib/split_csv.h"


#include "MyDb.hpp"
#include "def.h" 

using namespace std;

map < string, vector < string > >table2name;
map < string, vector < string > >table2type;
map < string, vector < string > >table2pkey;

//vector < string > result;


void
dumpKeyBuffer (ostream & os, _byte * buf, TableInfo & info)
{
  _field_offset offset = 0;
  os << "key: ";
  for (int i = 0; i < info.pKey_idx.size (); i++)
    {
      int idx = info.pKey_idx[i];
      if (info.colType[idx].type == INTEGER)
	{
	  os << (int) BL_CONV (*( int *) (buf + offset)) << "\t";	//conv to little endian
	  offset += 4;
	}
      else if (info.colType[idx].type == VARCHAR)
	{
	  string tmp ((char *) (buf + offset + sizeof (_field_offset)),
		      *(_field_offset *) (buf + offset));
	  os << tmp << '\t';
	  offset += tmp.length () + sizeof (_field_offset);
	}
      //offset += info.colType[idx].size;
    }
  os << endl;
}

_byte *
createKeyBuffer (_byte * dataBuf, TableInfo & info,
		 vector < _field_offset > &fo, _field_offset & keySize)
{
  _byte *buf = (_byte *) malloc (info.keySize);
  memset (buf, 0, info.keySize);
  _field_offset offset = 0;
  for (int i = 0; i < info.pKey_idx.size (); i++)
    {
      int idx = info.pKey_idx[i];
      if (info.colType[idx].type == INTEGER)
	{
	  //XXX,convert to big endian, BL_CONV any BUGS (untested for negative integer)??
	  *(unsigned int *) (buf + offset) = BL_CONV (*( int *) (dataBuf + fo[idx]));	//conv to big endian
	}
      else if (info.colType[idx].type == VARCHAR)
	{
	  memcpy (buf + offset, dataBuf + fo[idx], fo[i + 1] - fo[i]);
	}
      offset += fo[idx + 1] - fo[idx];
    }
  keySize = offset;
  return buf;
}

void
calcFieldOffset (const _byte * buf, const TableInfo * info,
		 vector < _field_offset > &fo)
{
  //cout<<"@@@"<<info->tableName<<endl;
  fo.resize (info->colType.size () + 1, 0);
  fo[0] = 0;
  _field_offset offset = 0;
  for (int i = 0; i < info->colType.size (); fo[++i] = offset)
    {
      //cout<<"!!! "<<offset<<endl;
      if (info->colType[i].type == INTEGER)
	offset += 4;
      else if (info->colType[i].type == VARCHAR)
	offset +=
	  (*(_field_offset *) (buf + offset)) + sizeof (_field_offset);
    }
  // cout<<offset<<"---- "<<endl;
}

_field_offset calcOneFieldOffset(const _byte * buf, const TableInfo * info, int n){
  _field_offset offset = 0;
  for(int i=0;i<n;i++){
      if (info->colType[i].type == INTEGER)
	offset += 4;
      else if (info->colType[i].type == VARCHAR)
	offset +=(*(_field_offset *) (buf + offset)) + sizeof (_field_offset);
  }
  return offset;
}

inline int
getInteger (const TableCol & tc, const _byte * buf,
	    vector < _field_offset > &fo)
{
  TableInfo *ti = table_infos[tc.tableID];
  int x = *(int *) (buf + fo[tc.colID]);
  return x;
}

inline const char* getStringBuf(const TableCol & tc, const _byte * buf,
	   vector < _field_offset > &fo, int &len){
  TableInfo *ti = table_infos[tc.tableID];
  char *ptr = (char *) (buf + fo[tc.colID] + sizeof (_field_offset));
  len =  *(_field_offset *) (buf + fo[tc.colID]);
  return ptr;
 
}

inline string
getString (const TableCol & tc, const _byte * buf,
	   vector < _field_offset > &fo)
{
  int len;
  const char *ptr=getStringBuf(tc,buf,fo, len);
  string x (ptr, len);
  return x;
}

void
dumpRowBuffer (ostream & os, const _byte * buf, const TableInfo & info)
{
  os << "Dump row buffer..." << endl;
  vector < _field_offset > fo;
  calcFieldOffset (buf, &info, fo);
  for (int i = 0; i < info.colName.size (); i++)
    {
      os << '+' << fo[i] << '\t' << info.colName[i] << '\t';
      if (info.colType[i].type == INTEGER)
	{
	  os << *(int *) (buf + fo[i]) << endl;
	}
      else if (info.colType[i].type == VARCHAR)
	{
	  string data ((char *) (buf + fo[i] + sizeof (_field_offset)),
		       *(_field_offset *) (buf + fo[i]));
	  os << data << endl;
	}
    }
  os << endl;

}

inline int
string2varchar (const string & str, _byte * buf)
{
  *(_field_offset *) (buf) = str.length ();
  memcpy (buf + sizeof (_field_offset), str.c_str (), str.length ());
  return sizeof (_field_offset) + str.length ();
}

void
createRowFromCSV (const string & row, TableInfo & info)
{
  vector < string > items;
  vector < _field_offset > fo (items.size () + 1, 0);
  split_csv (row.c_str (), items);
  assert (items.size () == info.colName.size ());
  _byte *buf = (_byte *) malloc (info.entrySize);
  memset (buf, 0, info.entrySize);
  _field_offset offset = 0;
  fo[0] = 0;
  for (int i = 0; i < info.colName.size (); fo[++i] = offset)
    {
      //cout<<"!!! "<<offset<<endl;
      if (info.colType[i].type == INTEGER)
	{
	  int data = atoi (items[i].c_str ());
	  *(int *) (buf + offset) = data;
	  offset += 4;
	}
      else if (info.colType[i].type == VARCHAR)
	{
	  string data (items[i], 1, items[i].length () - 2);	//remove quotes
	  offset += string2varchar (data, buf + offset);
	}
    }
  //dumpRowBuffer(cout,buf,info);
  _field_offset keySize;
  _byte *keyBuf = createKeyBuffer (buf, info, fo, keySize);
  Dbt key (keyBuf, keySize);
  Dbt data (buf, offset);

  int ret = info.db->put (NULL, &key, &data, DB_NOOVERWRITE);
  if (ret == DB_KEYEXIST)
    {
      info.db->err (ret, "Put failed because key  already exists");
    }
  info.row_count++;
  free (keyBuf);
  free (buf);
}

void
dumpTable (ostream & os, TableInfo & info)
{
  Dbc *cursorp;
  info.db->cursor (NULL, &cursorp, 0);
  assert (cursorp != NULL);
  int ret;
  Dbt key, data;
  os << "Dumping Table: " << info.tableName << endl;
  while ((ret = cursorp->get (&key, &data, DB_NEXT)) == 0)
    {
      _byte *dd = (_byte *) data.get_data ();
      dumpKeyBuffer (os, (_byte *) key.get_data (), info);
      dumpRowBuffer (os, dd, info);
    }
  cursorp->close ();
}



void
create (const string & table_name, const vector < string > &column_name,
	const vector < string > &column_type,
	const vector < string > &primary_key)
{
  table2name[table_name] = column_name;
  table2type[table_name] = column_type;
  table2pkey[table_name] = primary_key;

  TableInfo info (table_name, column_name, column_type, primary_key);
  table2info[table_name] = info;
  table_infos.push_back (&table2info[table_name]);
  table2info[table_name].db->ti = &table2info[table_name];
  // info.dump (cout);
}



void
load (const string & table_name, const vector < string > &row)
{

  map < string, TableInfo >::iterator info = table2info.find (table_name);
  if (info == table2info.end ())
    {
      cerr << "ERROR: Table not exists: " << table_name << endl;
      return;
    }
  for (int i = 0; i < row.size (); i++)
    {
      //fprintf(fout, "%s\n", row[i].c_str());
      //vector<string> cols;
      //split_csv(row[i].c_str(), cols);
      createRowFromCSV (row[i], info->second);
    }
//    dumpTable(cout, info->second);
}

void
preprocess ()
{
  /* I am too clever; I don't need it. */
}

inline void
matchKeyword (const string & kw, const char *str)
{
  if (kw.compare (str))
    {
      cerr << "ERROR: syntax error " << kw << "  expected: " << str << endl;
      exit (1);
    }
}

void
executeInsert (const string & sql)
{
  vector < string > token;
  tokenize (sql.c_str (), token);

  /*for(int i=0;i<token.size();i++){
     cout<<token[i]<<' ';
     }
     cout<<endl; */
  int ptr = 0;
  matchKeyword (token[ptr++], "INSERT");
  matchKeyword (token[ptr++], "INTO");
  string table_name = token[ptr++];
  matchKeyword (token[ptr++], "VALUES");
  map < string, TableInfo >::iterator iter = table2info.find (table_name);
  assert (iter != table2info.end ());
  while (ptr < token.size ())
    {
      string csv = "";
      matchKeyword (token[ptr++], "(");
      while (token[ptr].compare (")") != 0)
	{
	  csv += token[ptr++];
	}
      ptr++;			// ')'
      //insert
      createRowFromCSV (csv, iter->second);
      //cout<<csv<<endl;
      if (token[ptr].compare (";") == 0)
	break;
      matchKeyword (token[ptr++], ",");
    }


  //dumpTable (cout, iter->second);
}

void
parseCondition (CondInfo & ci, const string & op1, const string & opc,
		const string & op2)
{
  ci.op1_table = colname2tc[op1].tableID;
//  ci.op2_table = colname2tableID[op2];
  TableInfo *t1 = table_infos[ci.op1_table];
  assert (t1 != 0);
  //cerr<<"########"<<op1<<" "<<t1->tableName<<endl;
  ci.op1 = t1->name2idx[op1];
  ci.op2_table = -1;
  if (opc == ">")
    {
      ci.type = CT_GREATER;
      ci.int_op2 = atoi (op2.c_str ());
    }
  else if (opc == "<")
    {
      ci.type = CT_LESS;
      ci.int_op2 = atoi (op2.c_str ());
    }
  else if (opc == "=")
    {
      if (op2[0] == '\'')
	{			//string const
	  ci.type = CT_CONST_STRING;
	  ci.str_op2 = op2.substr (1, op2.length () - 2);
	}
      else if ((op2[0] >= '0' && op2[0] <= '9') || (op2[0] == '-'))
	{			//int const
	  ci.type = CT_CONST_INT;
	  ci.int_op2 = atoi (op2.c_str ());
	}
      else
	{			//join
	  cerr << "ERROR: JOIN unsupported" << endl;
	  ci.type = CT_JOIN;
	  ci.op2_table = colname2tc[op2].tableID;
	}
    }
  else
    {
      cerr << "ERROR: unknown condition: " << opc << endl;
    }
//  cout<<ci.type<<' '<<ci.str_op2<<endl;
}

vector < int >
TC2TableIdx (const vector < TableCol > &tc, const vector < int >&tables)
{
  vector < int >ret;
  for (int i = 0; i < tc.size (); i++)
    {
      for (int j = 0; j < tables.size (); j++)
	if (tables[j] == tc[i].tableID)
	  {
	    ret.push_back (j);
	    break;
	  }
    }
  return ret;
}




string
opCast (const vector < _byte * >&bufs,
	  const vector < TableCol > &cols, const vector < int >&tc2tb,
	  vector < vector< _field_offset > > &fos, vector < string > *out)
{
  assert (tc2tb.size () == cols.size ());
  string ret("");
  if (out)
    out->clear ();
  char tmp[64];
//  vector < vector < _field_offset > >fos (tables.size ());
//  for (int i = 0; i < tables.size (); i++)
//    {
//      calcFieldOffset (bufs[i], table_infos[tables[i]], fos[i]);
//    }
  for (int i = 0; i < cols.size (); i++)
    {
      TableInfo *ti = table_infos[cols[i].tableID];
      if (ti->colType[cols[i].colID].type == INTEGER)
	{
	  int x = getInteger (cols[i], bufs[tc2tb[i]], fos[tc2tb[i]]);
	  sprintf (tmp, "%d,", x);
	  if (out)
	    out->push_back (tmp);
	  ret.append(tmp);
	}
      else if (ti->colType[cols[i].colID].type == VARCHAR)
	{
	  int len;
	  const char* ptr = getStringBuf (cols[i], bufs[tc2tb[i]], fos[tc2tb[i]],len);
	  if (out)
	    out->push_back (string(ptr,len));
	  ret.append("'");
	  ret.append(ptr,len);
	  ret.append("',");
	}
  }
  if (ret.length () > 1)
	ret.erase(ret.end()-1);
  return ret;
}


static ProductState ps;

#define PN_BUFFER_MAX 1024
void
opProductInit (ProductState & s, vector < int >&tables,
	       vector < TableCol > &tc, vector <CondInfo> &condInfo)
{
  if (tables.empty ())
    {
      s.done = true;
      return;
    }
  while (s.cursors.size ())
    s.cursors.pop ();
  s.bufs.clear ();
  s.condInfo = condInfo;
  s.tables = tables;
  //sort the tables by their row_count can reduce overhead
  sort(s.tables.begin(),s.tables.end(),cmpTableInfoByIdx);
  s.tc2tb = TC2TableIdx (tc, s.tables);
  s.done = false;
  s.results.clear ();
  s.fo_buf.clear();
  s.tc = tc;
  s.dbs.clear();
  if(condInfo.empty()){
    for(int i=0;i<s.tables.size();i++){
    	s.dbs.push_back(table_infos[s.tables[i]]->db);
	}
  }else{
	//TODO WhERE
	for(int i=0;i<s.tables.size();i++){
	  bool found = false;
	  for(int j=0;j<s.condInfo.size();i++){
		MyDb * op1_db = table_infos[s.tables[i]]->colType[s.condInfo[j].op1].db;
	    if(s.tables[i]==s.condInfo[j].op1_table && op1_db){
/*	    	switch(s.condInfo[j].type){
            case CT_CONST_STRING:
		    	break;
    	    case CT_CONST_INT:
		    case CT_CONST_GREATER:
		    case CT_CONST_LESS:
		   case CT_JOIN:
			break;
		  default:
		}
*/   
		 s.dbs.push_back(op1_db);//use secondary db
		 found = true;
		 break;
	  }
     }
	 if(!found){
	   s.dbs.push_back(table_infos[s.tables[i]]->db); //use primary db
	 }
  }
  }	
  Dbc *cur;
//  table_infos[s.tables[0]]->db->cursor (NULL, &cur, 0);
  s.dbs[0]->cursor(NULL, &cur, 0);
  assert (cur != NULL);
  s.cursors.push (cur);
}


bool
opProductNext (ProductState & s, string & strout )
{
  if (s.done == true)
    return false;
  if (!s.results.empty ())
    {
      strout = s.results.back ();
      s.results.pop_back ();
      return true;
    }

  int ret;
  Dbt key, data;
  Dbc *cur;
  int count = 0;
  while (!s.cursors.empty ())
    {
      cur = s.cursors.top ();
      ret = cur->get (&key, &data, DB_NEXT);
      if (ret == 0)
	{
	  s.bufs.push_back ((_byte *) data.get_data ());
	  vector< _field_offset > fos;
      calcFieldOffset ((_byte*)data.get_data(), table_infos[s.tables[s.cursors.size()-1]],fos);
	  s.fo_buf.push_back(fos);
	  if (s.cursors.size () >= s.tables.size ())
	    {
	      //cout<<"DDD..."<<bufs.size()<<endl;
	      string retstr (opCast (s.bufs, s.tc, s.tc2tb, s.fo_buf, NULL));
	      //cout<<strout<<endl;
	      s.bufs.pop_back ();
		  s.fo_buf.pop_back();
	      s.results.push_back (retstr);
	      count++;
	      if (count >= PN_BUFFER_MAX)
    		break;
	    }
	  else
	    {
		  s.dbs[s.cursors.size()]->cursor(NULL, &cur, 0);
//	      table_infos[s.tables[s.cursors.size ()]]->db->cursor (NULL, &cur, 0);
	      assert (cur != NULL);
	      s.cursors.push (cur);
	    }
	}
      else
	{			//over
	  cur->close ();
	  s.cursors.pop ();
	  //XXX any bugs? e.g. empty table
	  if (s.bufs.size ()){
	    s.bufs.pop_back ();
		s.fo_buf.pop_back();
	  }
	}
    }
  //cout<<"#$#$ "<<count<<endl;
  if (!s.results.empty ())
    {
      strout = s.results.back ();
      s.results.pop_back ();
      return true;
    }
  else
    {
      s.done = true;
      return false;
    }
  //for(int i=0;i<tables.size();i++);
//  cout<<"COUNT="<<count<<endl;
}

void
opProduct_all (vector < int >&tables, vector < TableCol > &tc)
{
  if (tables.empty ())
    return;

  stack < Dbc * >cursors;
  Dbc *cur;
  table_infos[tables[0]]->db->cursor (NULL, &cur, 0);
  assert (cur != NULL);
  cursors.push (cur);
  int ret;
  Dbt key, data;
  vector < _byte * >bufs;
  vector < int >tc2tb (TC2TableIdx (tc, tables));
  int count = 0;
  while (!cursors.empty ())
    {
      cur = cursors.top ();
      ret = cur->get (&key, &data, DB_NEXT);
      if (ret == 0)
	{
	  bufs.push_back ((_byte *) data.get_data ());
	  if (cursors.size () >= tables.size ())
	    {
	      count++;
	      //cout<<"DDD..."<<bufs.size()<<endl;
	      vector < string > out;
	     // string strout = opCast (bufs, tc, tc2tb, tables, &out);
	     // cout << strout << endl;
	      bufs.pop_back ();
	    }
	  else
	    {
	      table_infos[tables[cursors.size ()]]->db->cursor (NULL,
									 &cur,
									 0);
	      assert (cur != NULL);
	      cursors.push (cur);
	    }
	}
      else
	{			//over
	  cur->close ();
	  cursors.pop ();
	  //XXX any bugs? e.g. empty table
	  if (bufs.size ())
	    bufs.pop_back ();
	}
    }
  //for(int i=0;i<tables.size();i++);
//  cout<<"COUNT="<<count<<endl;
}

void
parseSelect (const string & sql, vector < TableCol > &outputCol,
	     vector < CondInfo > &cond, vector < int >&table)
{
  outputCol.clear ();
  cond.clear ();
  int i;
  int ptr = 0;
  vector < string > token;
  tokenize (sql.c_str (), token);

  matchKeyword (token[ptr++], "SELECT");
  tokenize (sql.c_str (), token);
  for (i = 0; i < token.size (); i++)
    {
      if (token[i] == "SELECT" || token[i] == ",")
	continue;
      if (token[i] == "FROM")
	break;
      outputCol.push_back (colname2tc[token[i]]);
    }
  //TODO in fact, if no columns in a table after FROM are selected, we can ignore this table
  for (i++; i < token.size (); i++)
    {
      if (token[i] == "," || token[i] == ";")
	continue;
      if (token[i] == "WHERE")
	break;
      table.push_back (table2info[token[i]].tableID);
    }
  for (i++; i < token.size (); i++)
    {
      if (token[i] == ";" || token[i] == "AND")
	{
	  CondInfo tmp;
	  parseCondition (tmp, token[i - 3], token[i - 2], token[i - 1]);
	  cond.push_back (tmp);
	}
    }
}

void
executeSelect (const string & sql)
{

  vector < int >table;
  vector < TableCol > outputCol;
  vector < CondInfo > cond;

  parseSelect (sql, outputCol, cond, table);
  if (cond.empty ())
    {				//no condition
      //   opProduct_all(table, outputCol);
      opProductInit (ps, table, outputCol, cond);
    }
  else
    {
	  opProductInit( ps, table, outputCol, cond);
//      cerr << "ERROR: UNSUPPORT WHERE" << endl;
    }

}

void
execute (const string & sql)
{
  vector < string > token, output, table, row;
  map < string, int >m;
  int i;

  cout<<"RUNNING: "<<sql<<endl;
  //result.clear ();


  if (sql.compare (0, 6, "INSERT") == 0)
    {
      executeInsert (sql);
    }
  else if (sql.compare (0, 6, "SELECT") == 0)
    {
      executeSelect (sql);
    }
  else
    {
      fprintf (stderr, "Unsupported :%s\n", sql.c_str ());
      exit (1);
    }

}

int
next (char *row)
{
  vector < string > out;
  string strout;
  bool ret = opProductNext (ps, strout);
  if (ret == false)
    return (0);
  strcpy (row, strout.c_str ());

  /*
   * This is for debug only. You should avoid unnecessary output
   * when submitting, which will hurt the performance.
   */
//  printf ("%s\n", row);
  return (1);
}

int
get_key_integer (Db * sdbp,	// secondary db handle
	       const Dbt * pkey,	// primary db record's key
	       const Dbt * pdata,	// primary db record's data
	       Dbt * skey)	// secondary db record's key
{
  MyDb *db = (MyDb*)sdbp;
  // First, extract the structure contained in the primary's data
 // vector<_field_offset> fo;
  _byte *buf = (_byte*)pdata->get_data();
  _field_offset fo = calcOneFieldOffset(buf , db->ti, db->colIdx);
  // Now set the secondary key's data to be the representative's name

  //cout<<"### "<<db->ti->tableName<<" "<<db->colIdx<<" "<<fo<<endl;
  int *x = new int( *(int*)(buf+fo) );
  *x = BL_CONV((*x));
  skey->set_flags(DB_DBT_APPMALLOC);
  skey->set_data(x);
  skey->set_size(4);
  //cout<<(int)BL_CONV(*(int*)skey->get_data())<<endl;
  // Return 0 to indicate that the record can be created/updated.
  return (0);
}

int
get_key_string (Db * sdbp,	// secondary db handle
	       const Dbt * pkey,	// primary db record's key
	       const Dbt * pdata,	// primary db record's data
	       Dbt * skey)	// secondary db record's key
{
  MyDb *db = (MyDb*)sdbp;
  _byte *buf = (_byte*)pdata->get_data();
  _field_offset fo = calcOneFieldOffset(buf , db->ti, db->colIdx);
  //cout<<"### "<<db->ti->tableName<<" "<<db->colIdx<<" "<<fo<<endl;
  skey->set_data(buf+fo);
  _field_offset len = *(_field_offset*)(buf+fo);
  skey->set_size(sizeof(_field_offset)+len);
  //cout<<*(int*)skey->get_data()<<endl;
  // Return 0 to indicate that the record can be created/updated.
  return (0);
}

void
train (const vector < string > &query, const vector < double >&weight)
{
  /* I am too clever; I don't need it. */
  for (int i = 0; i < query.size (); i++)
    {
      const string & sql = query[i];
      if (sql.compare (0, 6, "SELECT") != 0 || weight[i] < 10.0)
	continue;
      vector < int >table;
      vector < TableCol > outputCol;
      vector < CondInfo > cond;
      parseSelect (sql, outputCol, cond, table);
      for (int i = 0; i < cond.size (); i++)
	{
	  TableInfo *ti1 = table_infos[cond[i].op1_table];
	  ColType & op1 = ti1->colType[cond[i].op1];
	  switch (cond[i].type)
	    {
	    case CT_CONST_STRING:
              ti1->createSecondaryDb(ti1->colName[cond[i].op1],get_key_string);
	      break;
	    case CT_GREATER:
	    case CT_LESS:
	    case CT_CONST_INT:
              ti1->createSecondaryDb(ti1->colName[cond[i].op1],get_key_integer);
	      break;
	    default:		//join
	      cerr << "TRAIN WARNING: unimplemented cond type:" << cond[i].
		type;
	    }
	}
    }

}

void
close ()
{
  /* I have nothing to do. */
}
